package com.jiaying.sync.kafka;

import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileReader;
import java.io.FileWriter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Kafka偏移量管理器
 * 负责Kafka消费者偏移量的保存、加载和管理
 */
public class OffsetManager {
    private static final Logger logger = LoggerFactory.getLogger(OffsetManager.class);
    private static final String OFFSET_FILE = "kafka_offsets.properties";
    private String offsetFilePath;
    private Map<String, Long> lastSavedOffsets = new ConcurrentHashMap<>();
    
    /**
     * 构造函数
     */
    public OffsetManager() {
        this.offsetFilePath = OFFSET_FILE;
    }
    
    /**
     * 构造函数
     * @param offsetFilePath 偏移量文件路径
     */
    public OffsetManager(String offsetFilePath) {
        this.offsetFilePath = offsetFilePath;
    }
    
    /**
     * 检查偏移量是否发生变化
     * @param currentOffsets 当前偏移量
     * @return 是否发生变化
     */
    public boolean hasOffsetChanged(Map<String, Long> currentOffsets) {
        if (lastSavedOffsets.size() != currentOffsets.size()) {
            return true;
        }
        
        for (Map.Entry<String, Long> entry : currentOffsets.entrySet()) {
            String key = entry.getKey();
            Long currentOffset = entry.getValue();
            Long lastOffset = lastSavedOffsets.get(key);
            
            if (lastOffset == null || !lastOffset.equals(currentOffset)) {
                return true;
            }
        }
        
        return false;
    }
    
    /**
     * 保存偏移量到文件
     * @param offsetMap 偏移量映射
     */
    public void saveOffsetsToFile(Map<String, Long> offsetMap) {
        try {
            java.util.Properties props = new java.util.Properties();
            for (Map.Entry<String, Long> entry : offsetMap.entrySet()) {
                props.setProperty(entry.getKey(), String.valueOf(entry.getValue()));
            }
            
            props.store(new FileWriter(offsetFilePath), 
                       "Kafka offsets saved at " + new Date());
            
            // 更新上次保存的偏移量
            lastSavedOffsets = new ConcurrentHashMap<>(offsetMap);
            logger.info("偏移量已保存到文件: {}", offsetFilePath);
        } catch (Exception e) {
            logger.error("保存偏移量到文件时出错: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 从文件加载偏移量
     * @return 偏移量映射
     */
    public Map<String, Long> loadOffsetsFromFile() {
        Map<String, Long> offsetMap = new ConcurrentHashMap<>();
        try {
            java.util.Properties props = new java.util.Properties();
            props.load(new FileReader(offsetFilePath));
            
            for (String key : props.stringPropertyNames()) {
                offsetMap.put(key, Long.parseLong(props.getProperty(key)));
            }
        } catch (Exception e) {
            logger.info("偏移量文件不存在或无法读取，将使用默认偏移量: {}", e.getMessage());
        }
        return offsetMap;
    }
    
    /**
     * 从文件加载偏移量并设置到分区
     * @param partitions 分区列表
     * @param seekFunction 设置偏移量的函数
     */
    public void loadOffsets(Collection<TopicPartition> partitions, OffsetSeekFunction seekFunction) {
        try {
            Map<String, Long> offsetMap = loadOffsetsFromFile();
            Map<TopicPartition, Long> partitionOffsets = new HashMap<>();
            
            for (TopicPartition partition : partitions) {
                String key = partition.topic() + "-" + partition.partition();
                if (offsetMap.containsKey(key)) {
                    partitionOffsets.put(partition, offsetMap.get(key));
                }
            }
            
            // 设置消费者偏移量
            for (Map.Entry<TopicPartition, Long> entry : partitionOffsets.entrySet()) {
                seekFunction.seek(entry.getKey(), entry.getValue());
            }
            
            // 更新上次保存的偏移量
            lastSavedOffsets = new ConcurrentHashMap<>(offsetMap);
            
            logger.info("已从文件加载偏移量: {}", offsetFilePath);
        } catch (Exception e) {
            logger.error("加载偏移量时出错: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 获取上次保存的偏移量
     * @return 上次保存的偏移量
     */
    public Map<String, Long> getLastSavedOffsets() {
        return new ConcurrentHashMap<>(lastSavedOffsets);
    }
    
    /**
     * 设置上次保存的偏移量
     * @param offsets 偏移量映射
     */
    public void setLastSavedOffsets(Map<String, Long> offsets) {
        this.lastSavedOffsets = new ConcurrentHashMap<>(offsets);
    }
    
    /**
     * 偏移量设置函数接口
     */
    @FunctionalInterface
    public interface OffsetSeekFunction {
        void seek(TopicPartition partition, long offset);
    }
}